package com.example.mq.rocketmq.consumer.ack;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * 推送模式的消费者
 * 对于消费端，如果我们在使用 Push 模式的情况下，只有我们消费返回成功，才会异步定期更新消费进度到 Broker 上。
 *
 *1.如果业务的回调没有处理好而抛出异常，会认为是消费失败当ConsumeConcurrentlyStatus.RECONSUME_LATER处理。
 *2.当使用顺序消费的回调MessageListenerOrderly时，由于顺序消费是要前者消费成功才能继续消费，所以没有RECONSUME_LATER的这个状态，只有SUSPEND_CURRENT_QUEUE_A_MOMENT来暂停队列的其余消费，直到原消息不断重试成功为止才能继续消费。
 *
 * @author luowj12
 * @since 2024/4/09 14:48
 */
public class MQPushACKConsumer {

    /**
     * 顺序消息消费，带事务方式（应用可控制Offset什么时候提交）
     */
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("sync_store_group");
        consumer.setNamesrvAddr("10.58.218.163:9876");
        /**
         * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
         * 如果非第一次启动，那么按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //订阅指定得topic和tag
        consumer.subscribe("secp_smc_topic", "tag_sync_store");
        //顺序消费  顺序消息消费的实现方式是，RocketMQ 会将同一个队列中的消息按照顺序传递给同一个消费者实例，确保消息在队列中的顺序性，从而保证消息被消费的顺序性
        consumer.registerMessageListener(new MessageListenerOrderly() {
            Random random = new Random();
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                context.setAutoCommit(true);
                System.out.println(msgs.size());
                for (MessageExt msg : msgs) {
                    //获取信息体
                    System.out.println(new String(msg.getBody()));
                    //转成实体
                    /*JSONArray objects = JSONUtil.parseArray(new String(msg.getBody()));
                    List<StoreDemo> storeSyncVOS = JSONUtil.toList(objects, StoreDemo.class);
                    System.out.println("==============>" + storeSyncVOS);*/
                }
                try {
                    //模拟业务处理失败的情况
                    int number = 1/0;
                } catch (Exception e) {
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        //并发消息消费的实现方式是，RocketMQ 会将同一个队列中的消息同时传递给多个消费者实例，并行处理消息，从而提高消息的处理速度和并发度
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            Random random = new Random();
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.println(msgs.size());
                for (MessageExt msg : msgs) {
                    //获取信息体
                    System.out.println(new String(msg.getBody()));
                    //转成实体
                    /*JSONArray objects = JSONUtil.parseArray(new String(msg.getBody()));
                    List<StoreDemo> storeSyncVOS = JSONUtil.toList(objects, StoreDemo.class);
                    System.out.println("==============>" + storeSyncVOS);*/
                }
                try {
                    //模拟业务处理失败的情况
                    int number = 1/0;
                } catch (Exception e) {
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.println("Consumer Started.");
    }

}
